#include "Debug.h"
#include "MFarm.h"

 MFarm::MFarm(DbParam dbParameters, const char *readers,const char *writers, bool debug)
: Farm(dbParameters)
{
        this->debug_ = debug;
	initilize(readers, writers);
}

bool MFarm::initilize(const char *readers,const char *writers)
{
        char **hosts = g_strsplit(readers, HOST_HOST_SPLIT, -1);
        int hosts_num =  g_strv_length(hosts);
        for(int i=0; i< hosts_num ; i++ )
        {
                char **hps = g_strsplit(hosts[i], HOST_PORT_SPLIT, -1);
                int  hps_num =  g_strv_length(hps);
                DbHost *db = NULL;
                if(hps_num == 1)
                {
                        db = new DbHost();
                        db->dbHost_ = g_strstrip(hps[0]);
                        db->dbPort_ = myDb_.dbPort_;
                }
                else if(hps_num == 2)
                {
                        db = new DbHost();
                        db->dbHost_ =g_strstrip(hps[0]);
                        db->dbPort_ = atoi(hps[1] );
                }
                else
                {
                   ERRO << hosts[i] <<  " is error! "<<endl;
                   abort();
		}
		if(db != NULL)
		{
			db->deadTime_ = 0;
			db->dbLive_ = true;
			readers_.push_back(db);
		}
                g_strfreev (hps);
        }
        g_strfreev (hosts );

        hosts = g_strsplit(writers, HOST_HOST_SPLIT, -1);
        hosts_num =  g_strv_length(hosts);
        for(int i=0; i< hosts_num ; i++ )
        {
                char **hps = g_strsplit(hosts[i], HOST_PORT_SPLIT, -1);
                int  hps_num =  g_strv_length(hps);
                PDbHost *db = NULL;
                if(hps_num == 1)
                {
                        db = new PDbHost();
                        db->dbHost_ = g_strstrip(hps[0]);
                        db->dbPort_ = myDb_.dbPort_;
                }
                else if(hps_num == 2)
                {
                        db = new PDbHost();
                        db->dbHost_ =g_strstrip(hps[0]);
                        db->dbPort_ = atoi(hps[1] );
                }
                else
                {
                   ERRO << hosts[i] <<  " is error! "<<endl;
                   abort();
		}
		if(db != NULL)
		{
			db->deadTime_ = 0;
			db->dbLive_ = true;
			PGconn *c = NULL;

			c = PQsetdbLogin(db->dbHost_.c_str(), db->dbPort_.c_str(),NULL, NULL, myDb_.dbName_.c_str(), myDb_.dbUser_.c_str(), myDb_.dbPass_.c_str());

			if(PQstatus(c) != CONNECTION_OK)
			{
				db->dbLive_ =  false;
				db->deadTime_ = time(NULL);
				if(debug_)
					cerr<<"Could not connect "<< myDb_.dbName_ <<", "<< db->dbHost_ << ", " << myDb_.dbUser_ << endl;
				ERRO << PQerrorMessage(c) << endl;
                                PQfinish(c);
                                c = NULL;
			}
                      	db->pDb = c;
			writers_.push_back(db);
		}
                g_strfreev (hps);
        }
        g_strfreev (hosts );
        rCount_ = 0;
        return true;
}

PGconn *MFarm:: GetWriter(int idx)
{
#ifdef _THREAD_SAFE
        pthread_mutex_lock(&_lock);
        DEBUG1("++++++++++++++     pthread_mutex_lock, please sure to use unlock");
#endif

	int size = (int)writers_.size();
	if( idx<0 || idx >= size )
	{
		error_ = "writer's idx is out bounder";
		return NULL;
	}

	PDbHost *dh = (PDbHost *) writers_[idx];
        if(dh->pDb == NULL)
	{
		dh->deadTime_ = 0;
		dh->dbLive_ = true;
		dh->pDb = PQsetdbLogin(dh->dbHost_.c_str(), dh->dbPort_.c_str(),NULL, NULL, myDb_.dbName_.c_str(), myDb_.dbUser_.c_str(), myDb_.dbPass_.c_str());
		if(PQstatus( dh->pDb ) != CONNECTION_OK)
		{
			dh->dbLive_ =  false;
			dh->deadTime_ = time(NULL);
			if(debug_)
				cerr<<"Could not connect "<< myDb_.dbName_ <<", "<< dh->dbHost_ << ", " << myDb_.dbUser_ << endl;
			ERRO << PQerrorMessage( dh->pDb ) << endl;
			PQfinish( dh->pDb );
			dh->pDb = NULL;
		}
                return dh->pDb;
        }  
        if(PQstatus( dh->pDb) == CONNECTION_OK)
                return dh->pDb;
        else
            ERRO <<"error : "<< PQerrorMessage(dh->pDb) << endl;

         if(PQstatus(dh->pDb) == CONNECTION_BAD)
               PQreset(dh->pDb);

        if(PQstatus( dh->pDb) == CONNECTION_OK)
                return dh->pDb;
        else
            ERRO <<"error : "<< PQerrorMessage(dh->pDb) << endl;

	return NULL;
}

PGconn* MFarm::GetReader()
{
#ifdef _THREAD_SAFE
        pthread_mutex_lock(&_lock);
        DEBUG1("++++++++++++++++++  pthread_mutex_lock, please sure to use unlock");
#endif

     int size = (int)readers_.size();
     if(size <= 0)
             return NULL;
     int num = 0;
     PGconn *c = NULL;
     while(num < size )
     {
        c = getNextReader(num, size);
        if(c != NULL)
           return c;
     }
     error_ = "can't connection all readers";
     return NULL;
}

PGconn* MFarm::getNextReader(int &num, int size )
{
       num++;
       rCount_ = rCount_% size;
       DbHost *dh = (DbHost *) readers_[ rCount_ ++ ];
       if(!dh->dbLive_)
       {
          long lt = time(NULL);
          if(dh->deadTime_ > lt - 300)
              return NULL;
          dh->dbLive_ = true;
       }
       try
       {
	       PGconn *c = NULL;
	       c = PQsetdbLogin(dh->dbHost_.c_str(), dh->dbPort_.c_str(), NULL, NULL, myDb_.dbName_.c_str(), myDb_.dbUser_.c_str(), myDb_.dbPass_.c_str());

	       if(PQstatus(c) != CONNECTION_OK)
	       {
		       dh->dbLive_ =  false;
		       dh->deadTime_ = time(NULL);
		       if(debug_)
			       cerr<<"Could not connect "<< myDb_.dbName_ <<", "<< dh->dbHost_ << ", " << myDb_.dbUser_ << endl;
		       ERRO << PQerrorMessage(c) << endl;
                       PQfinish(c);
                       c = NULL;
               }
               return c;
       }
       catch (exception& e)
       {
               DEBUG8("Could not connect reader " , myDb_.dbName_ , ", "
                               , dh->dbHost_ , ", " , myDb_.dbUser_ , ", " , e.what());
       }
       return NULL;
}

void MFarm::closeAll()
{
#ifdef _THREAD_SAFE
        pthread_mutex_lock(&_lock);
        DEBUG1("+++++++++++++  pthread_mutex_lock, please sure to use unlock");
#endif

	for (size_t i=0; i < writers_.size(); ++i)
	{      
		PDbHost *dh = (PDbHost *) writers_[ (int)i ];
		if(dh->pDb != NULL)
		{
                        DEBUG1("close writer");
			PQfinish( dh->pDb);
			dh->pDb = NULL;
		}
	}
#ifdef _THREAD_SAFE
        pthread_mutex_unlock(&_lock);
        DEBUG1("--------------    pthread_mutex_unlock");
#endif

}

void MFarm::CloseConnection(PGconn *c, bool write) 
{
	if(!write && c != NULL )
	{
		PQfinish(c);
		c = NULL;
	}
        else
        {
          DEBUG1("current conn is a writer::persistant, can't close it!");
        }
#ifdef _THREAD_SAFE
        pthread_mutex_unlock(&_lock);
        DEBUG1("------------------    pthread_mutex_unlock");
#endif

} 

void MFarm::Print()
{
#ifdef _THREAD_SAFE
        pthread_mutex_lock(&_lock);
        DEBUG1("++++++++++++++++  pthread_mutex_lock, please sure to use unlock");
#endif

  cout<<"MFarm writers ::"<<endl;
	for (size_t i=0; i < writers_.size(); ++i)
	{      
		PDbHost *dh = (PDbHost *) writers_[ (int)i ];
	        cout<< "w["<< i<<"] = "<< dh->dbHost_.c_str()<< ":" << dh->dbPort_ <<":"<< dh->deadTime_ <<":"<< dh->dbLive_<<endl;

	}

  cout<<"MFarm readers ::"<<endl;
	for (size_t i=0; i < readers_.size(); ++i)
	{      
		DbHost *dh = (DbHost *) readers_[ (int)i ];
	        cout<< "r["<< i<<"] = "<< dh->dbHost_.c_str()<< ":" << dh->dbPort_ <<":"<< dh->deadTime_ <<":"<< dh->dbLive_<<endl;

	}
#ifdef _THREAD_SAFE
        pthread_mutex_unlock(&_lock);
        DEBUG1("------------------  pthread_mutex_unlock");
#endif

}
